﻿namespace JESAI.DistributedLock.ZooKeeper;

using JESAI.DistributedLock.Internal;
using org.apache.zookeeper;
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

/// <summary>
/// 基于ZooKeeper共享锁配方的分布式读写器锁(https://zookeeper.apache.org/doc/current/recipes.html).
/// </summary>
public sealed partial class ZooKeeperDistributedReaderWriterLock : IInternalDistributedReaderWriterLock<ZooKeeperDistributedReaderWriterLockHandle>
{
    private const string ReadNodePrefix = "read-",
        WriteNodePrefix = "write-";

    private readonly ZooKeeperSynchronizationHelper _synchronizationHelper;

    /// <summary>
    /// 根据提供的<paramref name="pat"/>、<paramref name="connectionString"/>和<paramref names="options"/>构造一个新锁。
    /// 
    /// 如果指定了<paramref name="assumePathExists"/>，则节点将不会作为获取的一部分创建，也不会
    /// 发布后删除（默认为false）。
    /// </summary>
    public ZooKeeperDistributedReaderWriterLock(
        ZooKeeperPath path,
        string connectionString,
        bool assumePathExists = false,
        Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(path, assumePathExists: assumePathExists, connectionString, options)
    {
        if (path == default) { throw new ArgumentNullException(nameof(path)); }
        if (path == ZooKeeperPath.Root) { throw new ArgumentException("Cannot be the root", nameof(path)); }
    }

    /// <summary>
    /// 根据提供的<paramref name="name"”/>、<paramref name="connectionString"/>和<paramref names="options"/>构造一个新锁。
    /// 
    /// 锁的路径将是根目录“/”的父节点。如果<paramref name="name"/>不是有效的节点名称，则将对其进行转换以确保
    /// 有效性。
    /// </summary>
    public ZooKeeperDistributedReaderWriterLock(string name, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(ZooKeeperPath.Root, name, connectionString, options)
    {
    }

    /// <summary>
    ///基于提供的<paramref name="directoryPath"/>、<paramref name="name"/>和<paramref names="connectionString"/>构造一个新锁。
    /// 
    ///锁的路径将是<paramref name="directoryPath"/>的父节点。如果<paramref name="name"/>不是有效的节点名称，则将对其进行转换以确保
    ///有效性。
    /// </summary>
    public ZooKeeperDistributedReaderWriterLock(ZooKeeperPath directoryPath, string name, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? options = null)
        : this(
              (directoryPath == default ? throw new ArgumentNullException(nameof(directoryPath)) : directoryPath).GetChildNodePathWithSafeName(name),
              assumePathExists: false,
              connectionString,
              options)
    {
    }

    private ZooKeeperDistributedReaderWriterLock(ZooKeeperPath nodePath, bool assumePathExists, string connectionString, Action<ZooKeeperDistributedSynchronizationOptionsBuilder>? optionsBuilder) =>
        this._synchronizationHelper = new ZooKeeperSynchronizationHelper(nodePath, assumePathExists, connectionString, optionsBuilder);

    /// <summary>
    /// zookeeper服务节点路径
    /// </summary>
    public ZooKeeperPath Path => this._synchronizationHelper.Path;

    /// <summary>
    /// 实现<see cref="IDistributedReaderWriterLock.Name"/>。显式实现，以避免与以下事实混淆：这将在传入名称旁边包含前导“/”和基目录。
    /// </summary>
    string IDistributedReaderWriterLock.Name => this.Path.ToString();

    async ValueTask<ZooKeeperDistributedReaderWriterLockHandle?> IInternalDistributedReaderWriterLock<ZooKeeperDistributedReaderWriterLockHandle>.InternalTryAcquireAsync(TimeoutValue timeout, CancellationToken cancellationToken, bool isWrite)
    {
        var nodeHandleTask = isWrite
            ? this._synchronizationHelper.TryAcquireAsync(HasAcquiredWriteLock, WaitForWriteLockAcquiredOrChange, timeout, cancellationToken, WriteNodePrefix, alternateNodePrefix: ReadNodePrefix)
            : this._synchronizationHelper.TryAcquireAsync(HasAcquiredReadLock, WaitForReadLockAcquiredOrChange, timeout, cancellationToken, ReadNodePrefix, alternateNodePrefix: WriteNodePrefix);
        // 我们在这里被迫使用同步而非异步，因为ZooKeeperNetEx没有同步API
        var nodeHandle = await nodeHandleTask.AwaitSyncOverAsync().ConfigureAwait(false);

        return nodeHandle != null ? new ZooKeeperDistributedReaderWriterLockHandle(nodeHandle) : null;
    }

    private static bool HasAcquiredReadLock(ZooKeeperSynchronizationHelper.State state) =>
        // 如果前面没有写入者，我们就有阅读锁
        state.SortedChildren.TakeWhile(t => t.Path != state.EphemeralNodePath).All(t => t.Prefix != WriteNodePrefix);

    private static async Task<bool> WaitForReadLockAcquiredOrChange(ZooKeeper zooKeeper, ZooKeeperSynchronizationHelper.State state, Watcher watcher)
    {
        var nextLowestWriteNode = state.SortedChildren.TakeWhile(t => t.Path != state.EphemeralNodePath)
            .Last(t => t.Prefix == WriteNodePrefix);
        //如果下一个最低的写节点已经消失，那么等待就结束了。否则，让观察者继续关注
        //节点，以便在它更改时通知我们（在此之前我们无法获取锁）
        return await zooKeeper.existsAsync(nextLowestWriteNode.Path, watcher).ConfigureAwait(false) == null;
    }

    private static bool HasAcquiredWriteLock(ZooKeeperSynchronizationHelper.State state) =>
        state.SortedChildren[0].Path == state.EphemeralNodePath;

    private static async Task<bool> WaitForWriteLockAcquiredOrChange(ZooKeeper zooKeeper, ZooKeeperSynchronizationHelper.State state, Watcher watcher)
    {
        var ephemeralNodeIndex = Array.FindIndex(state.SortedChildren, t => t.Path == state.EphemeralNodePath);
        var nextLowestChildNode = state.SortedChildren[ephemeralNodeIndex - 1].Path;
        //如果下一个最低的子节点已经消失，那么等待就结束了。否则，让观察者继续关注
        //节点，以便在它更改时通知我们（在此之前我们无法获取锁）
        return await zooKeeper.existsAsync(nextLowestChildNode, watcher).ConfigureAwait(false) == null;
    }
}
